package watcher

import (
	"bufio"
	"context"
	"github.com/go-redis/redis/v8"
	"strconv"
	"strings"
	"time"
	"watcher/client/config"
)

type RedisWatcher struct {
	client   *redis.Client
	Done     chan struct{}
	Duration time.Duration
}

type RedisInfo map[string]string

func (watcher *RedisWatcher) Watch() <-chan *WatchData {
	dataChan := make(chan *WatchData, 10)
	go watcher.watch(dataChan, watcher.Duration)
	return dataChan
}

func (watcher *RedisWatcher) watch(ch chan<- *WatchData, duration time.Duration) {
	defer close(ch)
	ticker := time.NewTicker(duration)
	for t := range ticker.C {
		select {
		case <-watcher.Done:
			ticker.Stop()
			break
		default:
			func() {
				info, err := watcher.client.Info(context.Background()).Result()
				if err != nil {
					return
				}
				var rInfo = make(RedisInfo)
				parseRedisInfo(info, &rInfo)
				cc, _ := strconv.Atoi(rInfo["connected_clients"])
				um, _ := strconv.ParseFloat(rInfo["used_memory"], 64)
				bc, _ := strconv.Atoi(rInfo["blocked_clients"])
				ch <- &WatchData{
					Measurement: "redis",
					Time:        t,
					Fields: map[string]interface{}{
						"connected_clients": cc,
						"used_memory":       um,
						"blocked_clients":   bc,
					},
				}
			}()
		}
	}
}

func (watcher *RedisWatcher) Close() {
	close(watcher.Done)
}

func parseRedisInfo(info string, rInfo *RedisInfo) {
	r := strings.NewReader(info)
	reader := bufio.NewReader(r)
	for {
		line, _, err := reader.ReadLine()
		if err != nil {
			break
		}
		s := string(line)

		if strings.Contains(s, ":") {
			d := strings.Split(s, ":")
			(*rInfo)[d[0]] = d[1]
		} else {
			continue
		}
	}
}

func NewRedisWatcher(conf config.Conf) (*RedisWatcher, error) {
	return &RedisWatcher{
		client: redis.NewClient(&redis.Options{
			Addr: "localhost:6379",
		}),
		Done:     make(chan struct{}),
		Duration: conf.Duration,
	}, nil
}

var RedisRegister = func(conf config.Conf) (string, Watcher, error) {
	wt, err := NewRedisWatcher(conf)
	return "redis", wt, err
}
